package com.ylxin.reactor;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.util.function.Consumer;

/**
 * @author 29048
 * @version V1.0
 * @Project reactor
 * @Package com.ylxin.reactor
 * @Description: 发布者Publisher 生产者
 * Create by 杨立新 @Date 2022/9/21 13:58
 */
public class FluxTest {
    public static void main(String[] args) {
        //函数式编程
//        Flux.create(new Consumer<FluxSink<? extends Object>>() {
//            @Override
//            public void accept(FluxSink<?> fluxSink) {
//
//            }
//        });

        //创建了一个数据的生产者 发布者 被观察者,生产者如果没有订阅没有消费者就不会做任何事情，不生产数据
        Flux<Object> flux = Flux.create(fluxSink -> {
            System.err.println("FluxTest中的方法main执行了");
            System.err.println("main===>" + Thread.currentThread().getName());
            fluxSink.next("第一个数据").next("第二个数据").next("第三个数据").complete();
        });
//        flux.subscribe();  //忽略数据进行订阅，就是单纯想要方法运行。
//        flux.subscribe(System.out::println);  //通过consumer将每一个数据传过来进行处理。
        //不知道的参数就先new一下。
        flux.subscribe(new Subscriber<Object>() {

            //通过subscription.request() 方法来要数据
            private Subscription subscription;

            /**
             * 当我们的订阅者和发布者建立关系的时候执行
             * @param subscription  中间的上下文对象，订阅者通过这个对象向发布者要数据。
             */
            @Override
            public void onSubscribe(Subscription subscription) {
                System.err.println("FluxTest中的方法onSubscribe执行了");
                //通过Subscription来向发布者要数据，然后转给订阅者。
                this.subscription = subscription;  //赋值后就不为空
                //背压 每次要一个数据，
                subscription.request(1);  //存
            }

            /**
             * 每当收到一个数据的时候就会执行
             * @param o  o代表收到的数据
             */
            @Override
            public void onNext(Object o) {
                System.err.println("FluxTest中的方法onNext执行了--->" + o);
                //每当处理完成之后可以继续要下一个数据。
                //通过subscription.request() 方法来要数据，没有这个方法就创建一个成员变量来通过第一个来赋值。
                System.err.println("main===>" + Thread.currentThread().getName());
                subscription.request(1);
            }

            /**
             * 当发生错误的时候执行
             * @param throwable
             */
            @Override
            public void onError(Throwable throwable) {
                System.err.println("FluxTest中的方法onError执行了");
                throwable.getStackTrace();
            }

            /**
             * 当完成的时候执行
             */
            @Override
            public void onComplete() {
                System.err.println("FluxTest中的方法onComplete执行了");
            }
        });



    }
}
